package dataframe

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object DataFrame_SQLDemo_JsonContainsArray {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName("DataFrame_SQLDemo_JsonContainsArray")

    val spark: SparkSession = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    import spark.implicits._

    val reviewDF: DataFrame = spark
      .read.json("data/Electronics_5.json")
    reviewDF.createOrReplaceTempView("tb_review")

    val selectedDF: DataFrame = spark.sql(
      """
        |select asin, overall, reviewTime, reviewerID, reviewerName
        |from tb_review
        |where overall>=3
        |""".stripMargin)

    val selectedJSONArrayElementDF: Dataset[Row] = reviewDF
      .select($"asin", $"overall", $"helpful")
      .filter($"helpful".getItem(0) < 3)
  }

}
